use std::sync::Arc;
use chrono::{NaiveDateTime, Duration as ChronoDuration, TimeZone};
use crate::logical_plan::expr::agg::AggExpr;
use crate::logical_plan::series::Series;
use crate::physical_plan::{PhysicalPlanExpr, PhysicalPlanTrait, logicalplan_func_map_physicalplan};
use std::collections::{VecDeque, HashMap};
use crate::run_plan::RunPlan;
use crate::data_frame::{RealValue, ConfValue, ElementValue};
use crate::logical_plan::expr::func::{FuncArg, logical_plan_func_rule_ref, func_arg_collect, func_name_collect};
use crate::conf_parse::{parse, Ast};
use crate::conf_init::{MapResult, ResultErr};

mod agg;
pub mod func;

#[derive(Debug,Clone)]
pub enum Operator {
    Eq,
    NotEq,
    Lt,
    Le,
    Gt,
    Ge,
    And,
    Or,
}
impl Operator{
    pub fn to_func_name(&self)->&str{
        match self{
            Operator::Eq => {"eq"},
            Operator::NotEq => {"neq"},
            Operator::Lt => {"lt"},
            Operator::Le => {"le"},
            Operator::Gt => {"gt"},
            Operator::Ge => {"ge"},
            Operator::And => {"and"},
            Operator::Or => {"or"},
        }
    }
}

#[derive(Debug)]
pub enum LiteralValue{
    Null,
    Boolean(bool),
    Utf8(String),
    UInt8(u8),
    UInt16(u16),
    UInt32(u32),
    UInt64(u64),
    Int8(i8),
    Int16(i16),
    Int32(i32),
    Int64(i64),
    Float32(f32),
    Float64(f64),
    Range{
        left: u64,
        right: u64,
    },
    DateTime(NaiveDateTime),
    Duration(ChronoDuration),
}


fn binary_expr(l: LogicalPlanExpr, op: Operator, r: LogicalPlanExpr) -> LogicalPlanExpr {
    LogicalPlanExpr::Binary {
        left: Box::new(l),
        op: op,
        right: Box::new(r),
    }
}

#[derive(Debug,Clone)]
pub enum LogicalPlanExpr {
    Not(Box<LogicalPlanExpr>),
    List(Vec<Box<LogicalPlanExpr>>), //列表型参数，比如数组类型元素判断
    Binary{
        left: Box<LogicalPlanExpr>,
        op: Operator,
        right: Box<LogicalPlanExpr>,
    },//>/</=/!=/>=/<= 等一系列两个元素的关系判断表达式集合
    Column(ElementValue),//某一列
    Columns(Vec<ElementValue>),//某些列
    Literal(RealValue), //指定值类型的字面值
    LiteralConf(ConfValue),   //不指定值类型的字面值
    Func{                 //可嵌套的函数
        name: String,     //函数名称
        input: Vec<LogicalPlanExpr>, //处理函数的参数
    },
    Take{
        expr: Box<LogicalPlanExpr>,
        idx: Box<LogicalPlanExpr>,
    },//根据idx取值
    SortBy{
        expr: Box<LogicalPlanExpr>,
        by: Vec<LogicalPlanExpr>,
        reverse: Vec<bool>,
    },
    Agg(AggExpr),
    Window{
        function: Box<LogicalPlanExpr>,
        partition_by: Vec<LogicalPlanExpr>,
        order_by: Option<Box<LogicalPlanExpr>>,
    },
    Wildcard,//通配符
    Slice{
        input: Box<LogicalPlanExpr>,
        offset: Box<LogicalPlanExpr>,
        length: Box<LogicalPlanExpr>,
    },
    Count,
}

pub enum ApplyOptions{
    ApplyGroups,//不太明白跟下面的区别
    ApplyList,//把groups数据收集成list，再调用函数
    ApplyFlat,//在收集之前先调用函数
}
pub struct FunctionOptions{
    pub collect_groups: ApplyOptions,
    pub input_wildcard_expansion: bool,//输入的通配符是否展开
}

impl LogicalPlanExpr {
    //基于用户配置来构建LogicalPlanExpr，而不是其他LogicalPlanExpr
    pub fn col(args: &[FuncArg], pre_expr: &HashMap<String, LogicalPlanExpr>) -> MapResult<LogicalPlanExpr> {
        if args.len() == 1{
            let t = &args[0];
            if let FuncArg::Column(t) = t{
                return MapResult::Ok(LogicalPlanExpr::Column(ElementValue{element:t.clone()}));
            }else{
                return MapResult::Err(ResultErr::new(format!("col only accept FuncArg::Column")));
            }
        }else if args.len() > 1{
            let mut tmp = Vec::new();
            for t in args.iter(){
                if let FuncArg::Column(t) = t{
                    tmp.push(ElementValue{element:t.clone()});
                }else{
                    return MapResult::Err(ResultErr::new(format!("col only accept FuncArg::Column")));
                }
            }
            return MapResult::Ok(LogicalPlanExpr::Columns(tmp));
        }else{
            return MapResult::Err(ResultErr::new(format!("col need at least one arg")));
        }
    }
    pub fn literal(args: &[FuncArg], pre_expr: &HashMap<String, LogicalPlanExpr>) -> MapResult<LogicalPlanExpr> {
        if args.len() == 1{
            let t = &args[0];
            if let FuncArg::Literal(t) = t{
                return MapResult::Ok(LogicalPlanExpr::LiteralConf(ConfValue::new(t.to_string())));
            }else{
                return MapResult::Err(ResultErr::new(format!("literal only accept FuncArg::Literal")));
            }
        }else{
            return MapResult::Err(ResultErr::new(format!("literal need at least one arg")));
        }
    }

    //基于已有LogicalPlanExpr来构建复杂LogicalPlanExpr
    pub fn eq(args: &[FuncArg], pre_expr: &HashMap<String, LogicalPlanExpr>) -> MapResult<LogicalPlanExpr> {
        if args.len() == 2{
            let left = &args[0];
            let right = &args[1];
            let left = match left{
                FuncArg::OtherExpr(t) => {
                    if let Some(left) = pre_expr.get(t){
                        left.clone()
                    }else{
                        let err_desc = format!("eq left is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[0..1],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[0..1],pre_expr)?
                },
            };
            let right = match right{
                FuncArg::OtherExpr(t) => {
                    if let Some(right) = pre_expr.get(t){
                        right.clone()
                    }else{
                        let err_desc = format!("eq right is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[1..2],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[1..2],pre_expr)?
                },
            };

            return MapResult::Ok(binary_expr(left.clone(),Operator::Eq, right.clone()));
        }else{
            return MapResult::Err(ResultErr::new(format!("eq args len must eq 2, {:?}", args)));
        }
    }
    pub fn neq(args: &[FuncArg], pre_expr: &HashMap<String, LogicalPlanExpr>) -> MapResult<LogicalPlanExpr> {
        if args.len() == 2{
            let left = &args[0];
            let right = &args[1];
            let left = match left{
                FuncArg::OtherExpr(t) => {
                    if let Some(left) = pre_expr.get(t){
                        left.clone()
                    }else{
                        let err_desc = format!("neq left is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[0..1],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[0..1],pre_expr)?
                },
            };
            let right = match right{
                FuncArg::OtherExpr(t) => {
                    if let Some(right) = pre_expr.get(t){
                        right.clone()
                    }else{
                        let err_desc = format!("neq right is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[1..2],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[1..2],pre_expr)?
                },
            };

            return MapResult::Ok(binary_expr(left.clone(),Operator::NotEq, right.clone()));
        }else{
            return MapResult::Err(ResultErr::new(format!("neq args len must eq 2, {:?}", args)));
        }
    }
    pub fn lt(args: &[FuncArg], pre_expr: &HashMap<String, LogicalPlanExpr>) -> MapResult<LogicalPlanExpr> {
        if args.len() == 2{
            let left = &args[0];
            let right = &args[1];
            let left = match left{
                FuncArg::OtherExpr(t) => {
                    if let Some(left) = pre_expr.get(t){
                        left.clone()
                    }else{
                        let err_desc = format!("lt left is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[0..1],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[0..1],pre_expr)?
                },
            };
            let right = match right{
                FuncArg::OtherExpr(t) => {
                    if let Some(right) = pre_expr.get(t){
                        right.clone()
                    }else{
                        let err_desc = format!("lt right is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[1..2],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[1..2],pre_expr)?
                },
            };

            return MapResult::Ok(binary_expr(left.clone(),Operator::Lt, right.clone()));
        }else{
            return MapResult::Err(ResultErr::new(format!("lt args len must eq 2, {:?}", args)));
        }
    }
    pub fn le(args: &[FuncArg], pre_expr: &HashMap<String, LogicalPlanExpr>) -> MapResult<LogicalPlanExpr> {
        if args.len() == 2{
            let left = &args[0];
            let right = &args[1];
            let left = match left{
                FuncArg::OtherExpr(t) => {
                    if let Some(left) = pre_expr.get(t){
                        left.clone()
                    }else{
                        let err_desc = format!("le left is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[0..1],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[0..1],pre_expr)?
                },
            };
            let right = match right{
                FuncArg::OtherExpr(t) => {
                    if let Some(right) = pre_expr.get(t){
                        right.clone()
                    }else{
                        let err_desc = format!("le right is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[1..2],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[1..2],pre_expr)?
                },
            };

            return MapResult::Ok(binary_expr(left.clone(),Operator::Le, right.clone()));
        }else{
            return MapResult::Err(ResultErr::new(format!("le args len must eq 2, {:?}", args)));
        }
    }
    pub fn gt(args: &[FuncArg], pre_expr: &HashMap<String, LogicalPlanExpr>) -> MapResult<LogicalPlanExpr> {
        if args.len() == 2{
            let left = &args[0];
            let right = &args[1];
            let left = match left{
                FuncArg::OtherExpr(t) => {
                    if let Some(left) = pre_expr.get(t){
                        left.clone()
                    }else{
                        let err_desc = format!("gt left is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[0..1],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[0..1],pre_expr)?
                },
            };
            let right = match right{
                FuncArg::OtherExpr(t) => {
                    if let Some(right) = pre_expr.get(t){
                        right.clone()
                    }else{
                        let err_desc = format!("gt right is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[1..2],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[1..2],pre_expr)?
                },
            };

            return MapResult::Ok(binary_expr(left.clone(),Operator::Gt, right.clone()));
        }else{
            return MapResult::Err(ResultErr::new(format!("gt args len must eq 2, {:?}", args)));
        }
    }
    pub fn ge(args: &[FuncArg], pre_expr: &HashMap<String, LogicalPlanExpr>) -> MapResult<LogicalPlanExpr> {
        if args.len() == 2{
            let left = &args[0];
            let right = &args[1];
            let left = match left{
                FuncArg::OtherExpr(t) => {
                    if let Some(left) = pre_expr.get(t){
                        left.clone()
                    }else{
                        let err_desc = format!("ge left is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[0..1],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[0..1],pre_expr)?
                },
            };
            let right = match right{
                FuncArg::OtherExpr(t) => {
                    if let Some(right) = pre_expr.get(t){
                        right.clone()
                    }else{
                        let err_desc = format!("ge right is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[1..2],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[1..2],pre_expr)?
                },
            };

            return MapResult::Ok(binary_expr(left.clone(),Operator::Ge, right.clone()));
        }else{
            return MapResult::Err(ResultErr::new(format!("ge args len must eq 2, {:?}", args)));
        }
    }
    pub fn not(args: &[FuncArg], pre_expr: &HashMap<String, LogicalPlanExpr>) -> MapResult<LogicalPlanExpr> {
        if args.len() == 1{
            if let FuncArg::OtherExpr(left) = &args[0]{
                if let Some(left) = pre_expr.get(left){
                    return MapResult::Ok(LogicalPlanExpr::Not(Box::new(left.clone())));
                }else{
                    return MapResult::Err(ResultErr::new(format!("not left Expr not exist")));
                }
            }else{
                return MapResult::Err(ResultErr::new(format!("not left must be Expr")));
            }
        }else{
            return MapResult::Err(ResultErr::new(format!("not args len must eq 1")));
        }
    }
    pub fn and(args: &[FuncArg], pre_expr: &HashMap<String, LogicalPlanExpr>) -> MapResult<LogicalPlanExpr> {
        if args.len() == 2{
            let left = &args[0];
            let right = &args[1];
            let left = match left{
                FuncArg::OtherExpr(t) => {
                    if let Some(left) = pre_expr.get(t){
                        left.clone()
                    }else{
                        let err_desc = format!("and left is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[0..1],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[0..1],pre_expr)?
                },
            };
            let right = match right{
                FuncArg::OtherExpr(t) => {
                    if let Some(right) = pre_expr.get(t){
                        right.clone()
                    }else{
                        let err_desc = format!("and right is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[1..2],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[1..2],pre_expr)?
                },
            };

            return MapResult::Ok(binary_expr(left.clone(),Operator::And, right.clone()));
        }else{
            return MapResult::Err(ResultErr::new(format!("and args len must eq 2, {:?}", args)));
        }
    }
    pub fn or(args: &[FuncArg], pre_expr: &HashMap<String, LogicalPlanExpr>) -> MapResult<LogicalPlanExpr> {
        if args.len() == 2{
            let left = &args[0];
            let right = &args[1];
            let left = match left{
                FuncArg::OtherExpr(t) => {
                    if let Some(left) = pre_expr.get(t){
                        left.clone()
                    }else{
                        let err_desc = format!("or left is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[0..1],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[0..1],pre_expr)?
                },
            };
            let right = match right{
                FuncArg::OtherExpr(t) => {
                    if let Some(right) = pre_expr.get(t){
                        right.clone()
                    }else{
                        let err_desc = format!("or right is Expr, but not exist [{}]", t);
                        return MapResult::Err(ResultErr::new(err_desc));
                    }
                },
                FuncArg::Literal(t) => {
                    LogicalPlanExpr::literal(&args[1..2],pre_expr)?
                },
                FuncArg::Column(t) => {
                    LogicalPlanExpr::col(&args[1..2],pre_expr)?
                },
            };

            return MapResult::Ok(binary_expr(left.clone(),Operator::Or, right.clone()));
        }else{
            return MapResult::Err(ResultErr::new(format!("or args len must eq 2, {:?}", args)));
        }
    }
}

/*
三大类表达式：
1. filter类
2. 简单处理/判断函数类型
3. 复杂处理类

conf支持层层嵌套的方式来配置 LogicalPlanExpr 表达式:

filter类：
    支持((<a> and <b>) || (<c> or <d>)) && <e>复杂灵活风格的配置
简单函数类：
    优先从LogicalPlanExpr方法中去匹配函数，如果没找到，就调用logicalplan_func_map_physicalplan
    从PhysicalPlan里面去找函数。都没找到就是不支持。

column("port") column_port
lit("80")      port_value_80
gt(<column_port>,<port_value_80>)  my_gt
*/

pub fn conf_func_map_logicalplan_expr(conf: &str,
                                      pre_expr: &HashMap<String, LogicalPlanExpr>)
                                      ->MapResult<LogicalPlanExpr>{

    let func_expr = logical_plan_func_rule_ref();
    let func = parse(conf, func_expr, false)?;

    if let Ast::Rule((_,func)) = func{
        if func.len() == 4{
            let func_name = &func[0];
            let func_arg = &func[2];

            let func_name = func_name_collect(func_name)?;
            let func_arg = func_arg_collect(func_arg)?;

            match func_name.as_str(){
                "col" => {return LogicalPlanExpr::col(func_arg.as_slice(),pre_expr);},
                "literal" => {return LogicalPlanExpr::literal(func_arg.as_slice(),pre_expr);},
                "eq" => {return LogicalPlanExpr::eq(func_arg.as_slice(),pre_expr);},
                "neq" => {return LogicalPlanExpr::neq(func_arg.as_slice(),pre_expr);},
                "lt" => {return LogicalPlanExpr::lt(func_arg.as_slice(),pre_expr);},
                "le" => {return LogicalPlanExpr::le(func_arg.as_slice(),pre_expr);},
                "gt" => {return LogicalPlanExpr::gt(func_arg.as_slice(),pre_expr);},
                "ge" => {return LogicalPlanExpr::ge(func_arg.as_slice(),pre_expr);},
                "not" => {return LogicalPlanExpr::not(func_arg.as_slice(),pre_expr);},
                "and" => {return LogicalPlanExpr::and(func_arg.as_slice(),pre_expr);},
                "or" => {return LogicalPlanExpr::or(func_arg.as_slice(),pre_expr);},

                other => {
                    return logical_plan_direct_map_physical_plan(func_name.as_str(), func_arg.as_slice(), pre_expr);
                },
            }
        }else{
            return MapResult::Err(ResultErr::new(format!("conf func map: func len must eq 4")));
        }
    }else{
        return MapResult::Err(ResultErr::new(format!("conf func map: Ast must be Rule")));
    }
}

fn logical_plan_direct_map_physical_plan(func_name: &str,
        args: &[FuncArg], pre_expr: &HashMap<String, LogicalPlanExpr>)->MapResult<LogicalPlanExpr>{

    let func = logicalplan_func_map_physicalplan(func_name)?;
    let mut logical_plan_expr = Vec::new();
    for t in args.iter(){
        logical_plan_expr.push(t.to_logical_plan(pre_expr)?);
    }

    return MapResult::Ok(LogicalPlanExpr::Func {name: func_name.to_string(), input: logical_plan_expr});
}









